package cnblogs.guzb.multikakfa;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author 顾志兵
 * @mail ipiger@163.com
 * @since 2024-03-23
 */
@EnableKafka
@Configuration
public class KafkaConfiguration {

    @Primary
    @Bean("consumerFactory")
    public ConsumerFactory createConsumerFactory() {
        Map<String, Object> consumerProperties = new HashMap<>();
        consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    // 第二个消费工厂，为便于实操, 这里依然连接的是同一个 Kafka 服务器
    @Bean("consumerFactory2")
    public ConsumerFactory createConsumerFactory2() {
        Map<String, Object> consumerProperties = new HashMap<>();
        consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    @Primary
    // 自己创建的监听容器工厂实例中，一定要有一个实例的名字叫: kafkaListenerContainerFactory,
    // 因为 KafkaAnnotationDrivenConfiguration 中也默认配置了一个 KafkaListenerContainerFactory，
    // 这个默认的 KafkaListenerContainerFactory 名称就叫 kafkaListenerContainerFactory，
    // 其装配条件就是当容器中没有名称为 kafkaListenerContainerFactory 的Bean时，那个装配就生效，
    // 如果不阻止这个默认的KafkaListenerContainerFactory装备，会导致容器中有两个 KafkaListenerContainerFactory，这会引入一些初始化问题
    @Bean("kafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<KafkaMessageListenerContainer> createContainerFactory1(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            @Qualifier("consumerFactory") ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        configurer.configure(listenerContainerFactory, consumerFactory);
        return listenerContainerFactory;
    }

    // 第二个监听器容器工厂
    @Bean("kafkaListenerContainerFactory2")
    public KafkaListenerContainerFactory<KafkaMessageListenerContainer> createContainerFactory2(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            @Qualifier("consumerFactory2") ConsumerFactory consumerFactory2) {
        ConcurrentKafkaListenerContainerFactory listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory();
        configurer.configure(listenerContainerFactory, consumerFactory2);
        return listenerContainerFactory;
    }

}
